package com.atguigu.Flink.timeAndwindow;

import com.atguigu.Flink.POJO.Event;
import com.atguigu.Flink.POJO.WordCount;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

public class Flink10_LateData {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<Event> ds = env.socketTextStream("hadoop102", 8888)
                .map(
                        line -> {
                            String[] split = line.split(",");
                            return new Event(split[0], split[1], Long.valueOf(split[2]));
                        }
                ).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                                .withTimestampAssigner(
                                        new SerializableTimestampAssigner<Event>() {
                                            @Override
                                            public long extractTimestamp(Event event, long l) {
                                                return event.getTs();
                                            }
                                        }
                                )
                );
        //侧输出流标签
        OutputTag<WordCount> outputTag = new OutputTag<>("id", Types.POJO(WordCount.class));

        ds.print("input");
        SingleOutputStreamOperator<WordCount> sum = ds.map(
                        event -> new WordCount(event.getUrl(), 1L)
                ).keyBy(
                        k -> k.getWord()
                ).window(
                        TumblingEventTimeWindows.of(Time.seconds(10))
                ).allowedLateness(Time.seconds(10)) //窗口延迟10s关闭,这个10s指的是时间时间的10秒哦！
                .sideOutputLateData(outputTag)//将最终迟到的数据发往侧流
                .sum("count");
        sum.print();

        //捕获侧输出流
        sum.getSideOutput(outputTag).print("侧流");



        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
